package com.shujia.streeam

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo6DriverHA {
  def main(args: Array[String]): Unit = {


    val conf = new SparkConf()
      //.setMaster("local[2]")
      .setAppName("Demo1WordCount")

    val sc = new SparkContext(conf)

    val checkpoint = "spark/data/driverha"

    /**
      *
      * 创建 streamming上下文对象  需要指定间隔时间   延迟时间
      *
      */

    val createContext = () => {

      println("初始化StreamingContext================")

      val ssc = new StreamingContext(sc, Durations.seconds(5))

      //设置数据备份路径
      ssc.checkpoint(checkpoint)

      val ds = ssc.socketTextStream("node1", 8888)

      ds.print()


      //返回上下文对象
      ssc
    }


    /**
      * getOrCreate  如果checkpoint目录里面已经有Driver的信息 那么直接中目录中恢复一个Driver
      * 如果checkpoint 目录不存在   再调用后面的createContext 方法创建Driver
      */
    val ssc = StreamingContext.getOrCreate(checkpoint, createContext)


    //启动ssc
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }
}
